#ifndef THREADPOOL_H
#define THREADPOOL_H
#include <vector>
#include <queue>
#include <mutex>
#include <iostream>
#include <functional>
#include <condition_variable>
#include <memory>
#include <atomic>
#include <chrono>
#include <thread>
#include <unordered_map>
#include <future>
#include <cassert>

#define TASK_THRESHOLD 1024
#define THREAD_THRESHOLD 20
#define THREAD_IDLETIME 10

using std::endl;
using std::cout;

using TaskFunc = std::function<void(int)>;
enum class ThreadPoolMode
{
    FIXED,
    CACHED
};

class Any
{
public:
    Any():_content(nullptr){}

    template<class T>
    Any(const T& val)
    :_content(new placeholders<T>(val))      //用子类的指针初始化父类的指针
    {}

    Any(const Any& any)
    :_content(any._content == nullptr ? nullptr : (any._content->clone()))      //指针使用注意要判空！
    {}

    Any& swap(Any any)      
    {
        std::swap(_content, any._content);
        return *this;
    }

    template<class T>
    Any& operator=(const T& val)    
    {
        swap(Any(val));
        return *this;       //可以直接return swap
    }    

    Any& operator=(Any& any)
    {
        swap(any);
        return *this;
    }

    ~Any()
    {
        if(_content != nullptr)
        delete _content;
    }

    template<class T>
    T* get()    
    {
        assert(typeid(T) == _content->type());
        return &(((placeholders<T>*)(_content)) ->_val);
    }

    class holders
    {
    public:
        virtual const std::type_info& type(){}   
        virtual holders* clone(){}   
        virtual ~holders(){}
    };

    template<class T>
    class placeholders : public holders
    {
    public:
        placeholders(T val)
        :_val(val)
        {}

        const std::type_info& type(){return typeid(T);}

        holders* clone()
        {
            return (new placeholders<T>(_val));
        }

    public:
        T _val;
    };

private:
    holders* _content;
};

//实现任务完成和获取结果之间的同步
class Semaphore
{
public:
    Semaphore():isExist_(true),limits_(0){}
    ~Semaphore()
    {
        cout << "debug ~Semaphore" << endl;
        isExist_ = false;
    }
    
    void wait()
    {
        if(!isExist_)
        {
            return;
        }

        std::unique_lock<std::mutex> lck(mutex_);
        cond_.wait(lck,[&](){return limits_ > 0;});
        limits_--;
    }

    void post()
    {
        if(!isExist_)
        {
            return;
        }

        cout << "debug: post start " << endl;
        std::unique_lock<std::mutex> lck(mutex_);
        
        limits_++;
        cond_.notify_all();     //需要手动唤醒！在这导致死锁了? --> 需要条件变量上确实有线程在wait --> 没有正确处理析构
        
        cout << "debug: post end " << endl;
    }

private:
    std::mutex mutex_;
    std::atomic_bool isExist_;
    std::condition_variable cond_;
    int limits_;
};

//抽象类
class Result;

class Task
{
public:
    Task(/*bool isNeedResult*/):result_(nullptr)/*, isNeedResult_(isNeedResult)*/{}
    virtual ~Task(){}
    void runAndSetResult();      //注意一种情况：任务执行完了，但Result还没构造完，这两也有同步关系
    virtual Any run() = 0;
    
    void setResult(Result* result)
    {
        result_ = result;
    }

    // bool isNeedResult()
    // {
    //     return isNeedResult_;
    // }

private:
    Result* result_;
   //bool isNeedResult_;     //明确 “要不要” 获取结果
};

class Result
{
public:
    Result(bool isValid = false, std::shared_ptr<Task> task = nullptr) 
    : isValid_(isValid), task_(task)
    {
        task_.get()->setResult(this);
    }  
    Result(){}
    Result(const Result& result) {}
    Result& operator=(Result& result) = default;

    //1.任务没完成，主线程先调用get，notify_all正常

    //2.任务先完成，主线程后调用get，notify_all也能正常
    //3.任务先完成，但主线程不调用get, notify_all阻塞
    //解决2、3关键要明确好 “要不要” 获取任务
    template<typename T>
    T* getVal()
    {   
        assert(/*!task_.get()->isNeedResult() &&*/ isValid_);

        // if(task_.get()->isNeedResult() == false)
        // {
        //     cout << "The task doesn't need a result!" << endl;
        //     return nullptr;
        // }
        // if(isValid_ == false)
        // {
        //     cout << "The Result is inValid!" << endl;
        //     return nullptr;
        // }
        
        semaphore_.wait();
        return any_.get<T>();
    }

    void setVal(const Any& data)
    {
        any_ = std::move(data);

        //if(task_.get()->isNeedResult())
        //{
            semaphore_.post(); 
        //}
    }
private:
    Any any_;

    bool isValid_;
    std::shared_ptr<Task> task_;    //一个任务对应一个结果

    Semaphore semaphore_;
};

void Task::runAndSetResult()
{
    //轮询，直到Result构建完成（因为Result会在构造函数中对task调用setResult）? 这里可以不用担心消耗资源，因为虽然有前后关系，但时间上差的不多
    if(result_ != nullptr)      
    {
        result_->setVal(run());
    }         
}


class Thread
{
public:
    Thread(const TaskFunc& taskFunc = nullptr): threadId_(generateId_++), taskFunc_(taskFunc){}
    void start()    //此时真正创建系统的线程
    {
        if(taskFunc_ == nullptr)
        {
            cout << "The taskFunc is invalid! " << endl;
            return;
        }
        
        std::thread t(taskFunc_, threadId_);
        t.detach();
    }

    int getId()
    {
        return threadId_;
    }

private:
    int threadId_;      
    //std::unique_ptr<std::thread> ptrThread_;      直接使用函数对象作为成员变量即可，系统线程在start方法中单独创建即可
    static int generateId_;
    TaskFunc taskFunc_;     
};

int Thread::generateId_ = 1;

class ThreadPool
{
public:
    ThreadPool()
    :mode_(ThreadPoolMode::FIXED),
    initialThreadCount_(4),
    threadThresHold_(THREAD_THRESHOLD),
    currentThreadCount_(0),
    taskThresHold_(TASK_THRESHOLD),
    currentTaskCount_(0),
    isPoolRunning_(false)
    {}
    
    ~ThreadPool()
    {
        while(currentTaskCount_ != 0);
        
        isPoolRunning_ = false;
        cout << "debug : ~ThreadPool start" << endl;

        std::unique_lock<std::mutex> lck(queueMutex_);
        notEmpty_.notify_all();
        exitCond_.wait(lck,[&](){return getCurrentThreadCount() == 0;});

        cout << "debug : ~ThreadPool end" << endl;
        
    }
    
    ThreadPool(const ThreadPool& pool) = delete;
    ThreadPool& operator=(const ThreadPool& pool) = delete;

    void close()
    {
        isPoolRunning_ = false;
    }
    int getCurrentThreadCount()
    {
        return currentThreadCount_;
    }

    void setMode(ThreadPoolMode mode)
    {
        if(isRunning())
        {
            cout << "Pool is Running!, please stop it first and try again" << endl;
            return;
        }
        mode_ = mode;
    }
    void setThreadThresHold(int threshold)
    {
        if(isRunning())
        {
            cout << "Pool is Running!, please stop it first and try again" << endl;
            return;
        }
        threadThresHold_ = threshold;
    }
    void setTaskThresHold(int threshold)
    {
        if(isRunning())
        {
            cout << "Pool is Running!, please stop it first and try again" << endl;
            return;
        }
        taskThresHold_ = threshold;
    }

    Result submitTask(std::shared_ptr<Task> task)
    {
        std::unique_lock<std::mutex>lck (queueMutex_);
        if(task == nullptr)
        {
            return Result(false, nullptr);
        }
        //两种阻塞的情况：
        //1.其他线程太忙，任务无法得到有效处理 --> 通过同步，阻塞用户对结果的的获取；
        //2.任务数量到达阈值 --> 一定时间后若仍未获取到锁，则“降级”服务，直接返回；

        while(taskQueue_.size() >= taskThresHold_)
        {
            cout << "debug : submitTask wait..." << endl;
            auto status = notFull_.wait_for(lck, std::chrono::seconds(1));
            if(status == std::cv_status::timeout)
            {
                return Result(false, task);
            }
        }
        taskQueue_.push(task);
        currentTaskCount_++;
        notEmpty_.notify_all();

        if(mode_ == ThreadPoolMode::CACHED && currentThreadCount_ < threadThresHold_ && currentThreadCount_ < currentTaskCount_)
        {
            //创建线程
            createAndRunThread(1);      //?：新增线程数量的策略
        }        
        return Result(true, task);
    }

    void start(int initialThreadCount)
    {
        isPoolRunning_ = true;
        if(initialThreadCount > threadThresHold_)       //超过阈值
        {
            //log
            return;
        }
        
        initialThreadCount_ = initialThreadCount;
        idleThreadCount_ = initialThreadCount;

        createAndRunThread(initialThreadCount);
    }

private:
    void createAndRunThread(int count)
    {
        for(int i = 0; i < count; i++)
        {
            Thread* t = new Thread(std::bind(&ThreadPool::threadFunc, this, std::placeholders::_1));
            threads_.emplace(t->getId(), std::unique_ptr<Thread>(t));
            cout << "debug: created " << t->getId() << " thread" << endl;
            t->start();
        } 
        idleThreadCount_ += count;
        currentThreadCount_ += count;
    }

    void eraseThread(int threadId)
    {
        cout << "debug: ready to erase " << threadId << endl;
        if(threads_.count(threadId) == 1)
        {
            threads_.erase(threadId);
            currentThreadCount_--;
        }
        else
        {
            cout << "The thread has been erased!" << endl;
        }

        exitCond_.notify_all();     //每次删除一个线程后都唤醒一下主线程看是否可以析构了?
    }

    bool isRunning()
    {
        return isPoolRunning_;
    }

    void threadFunc(int threadId)   //任务无需作为参数传递，直接从队中获取即可
    {
        //thread_local int threadLocalId = threadId;
        while(isRunning())
        {
            //cout << "debug: " << "this is thread " << threadId << endl; 
            std::shared_ptr<Task> task;
            {
                std::unique_lock<std::mutex> lck(queueMutex_);
                while(currentTaskCount_ == 0)         //没有任务时对应两种情况：1.确实没来得及提交 2.线程池已退出，后续不会再提交
                {
                    cout << "debug: thread " << threadId << " wait for task..." << endl;
                    if(!isRunning())
                    {
                        //notEmpty_.notify_all();
                        eraseThread(threadId);
                        return;      //出线程函数后，线程自动销毁
                    }             

                    if(mode_ == ThreadPoolMode::CACHED)
                    {
                        auto status = notEmpty_.wait_for(lck, std::chrono::seconds(THREAD_IDLETIME));
                        //被唤醒且抢到锁了，就一定不会超时
                        if(status == std::cv_status::timeout && currentThreadCount_ > initialThreadCount_)
                        {
                            //回收空闲线程
                            eraseThread(threadId);
                            return;     //出线程函数后，线程自动销毁
                        }
                    }
                    else    //fixed模式下，只等待被唤醒即可
                    {
                        notEmpty_.wait(lck);
                    }
                }

                idleThreadCount_--;     //线程拿到任务就一定会去执行，所以空闲线程数量--
                task = taskQueue_.front();            //从任务队列中获取任务
                taskQueue_.pop();
                currentTaskCount_--;
                cout << "debug: thread " << threadId << " got the task..." << endl;
                //先通知
                if(currentTaskCount_ > 0)
                {
                    notEmpty_.notify_all();
                }
                notFull_.notify_all();
            }//再释放锁

            if(task != nullptr)
            {
                task.get()->runAndSetResult();     //执行任务  
                cout << "debug: thread " << threadId << " finished the task..." << endl;                
            }

            idleThreadCount_++;
            //任务执行完后线程数量++，由于是atomic_int，所以无需担心线程安全
        }

        //notEmpty_.notify_all();
        eraseThread(threadId);
    }

    ThreadPoolMode mode_;

    int initialThreadCount_;
    int threadThresHold_;       //线程数量阈值
    std::atomic_int idleThreadCount_;   //空闲线程数量
    std::atomic_int currentThreadCount_;     //当前线程数量
    std::unordered_map<int, std::unique_ptr<Thread>> threads_;

    std::queue<std::shared_ptr<Task>> taskQueue_;
    int taskThresHold_;     //任务阈值
    std::atomic_int currentTaskCount_;     //不用.size()而用atomic_int，因为STL不是线程安全的，要保证线程安全
    
    std::mutex queueMutex_;
    std::condition_variable notFull_;
    std::condition_variable notEmpty_;
    std::condition_variable exitCond_;

    std::atomic_bool isPoolRunning_;
};


#endif